package cn.genmer.test.security.io.testreactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class SelectorThread extends ThreadLocal<LinkedBlockingQueue<Channel>> implements Runnable {
    // 每个线程对应一个selector
    // 多线程情况下,该主机,该程序的并发客户端被分配到多个selector上
    // 注意, 每个客户端,只绑定其中一个select

    Selector selector = null;
    LinkedBlockingQueue<Channel> lbq = get();
    SelectorThreadGroup stg;

    protected LinkedBlockingQueue<Channel> initialValue(){
        return new LinkedBlockingQueue<>();
    }

    SelectorThread(SelectorThreadGroup stg){
        try {
            this.stg = stg;
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    void setWorker(SelectorThreadGroup worker){
        stg.workerGroup = worker;
    }

    @Override
    public void run() {
        System.out.println("【线程组EventLoop】-" + Thread.currentThread().getName() + "启动! " + "当前事件数量：" + selector.keys().size());
        while (true){
            try {
                // // step 1 select 找多路复用器要东西,看看有没有事件注册
//                System.out.println("【EventLoop-"+ Thread.currentThread().getName() +"】-" + " select()前， " + "当前事件数量：" + selector.keys().size());
                int nums = selector.select(); // 阻塞

                // 多线程原因，跑太快这里暂停下可以达到正常的效果，不过这绝对不是解决方案
                // Thread.sleep(1000);
//                System.out.println("【EventLoop-"+ Thread.currentThread().getName() +"】-" + " select()后， " + "当前事件数量：" + selector.keys().size());
                // step 2 processSelectedKeys
                if (nums>0){
                    // 以下处理过程是线性的
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = keys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey next = iterator.next();
                        iterator.remove();
                        if (next.isAcceptable()){
                            // 比较复杂的地方: 接受客户端的过程 - 接收后注册, 多线程下,新客户端注册到哪里
                            acceptHandler(next);
                        } else if (next.isReadable()){
                            readHandler(next);
                        } else if (next.isWritable()){

                        }
                    }
                }
                // step 3 runAllTasks
                if (!lbq.isEmpty()){
                    // 本地变量是线程隔离的， 但是对象是堆中
                    Channel c = lbq.take();
                    if (c instanceof ServerSocketChannel){
                        ServerSocketChannel  server = (ServerSocketChannel)c;
                        server.register(selector, SelectionKey.OP_ACCEPT);
                        System.out.println(Thread.currentThread().getName()+" 监听 注册！");
                    } else if (c instanceof SocketChannel){
                        SocketChannel  client = (SocketChannel)c;
                        ByteBuffer buffer = ByteBuffer.allocate(4096);
                        client.register(selector, SelectionKey.OP_READ, buffer);
                    }
                }
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    private void readHandler(SelectionKey key){
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        SocketChannel client = (SocketChannel) key.channel();
        buffer.clear();
        while (true){
            try {
                int nums = client.read(buffer);
                if (nums>0){
                    buffer.flip(); // 将读到的内容翻转, 直接写出
                    if (buffer.hasRemaining()){
                        System.out.println("-------------------------------------------");
                        System.out.println(Thread.currentThread().getName() + " 来消息了！ from ：" + client.getRemoteAddress());
                        System.out.println("-------------------------------------------");
                        printBufferMessage(buffer);
                        client.write(buffer);
                    }
                    buffer.clear();
                } else if (nums == 0){
                    break;
                } else if (nums < 0){
                    System.out.println(client.getRemoteAddress() + "断开啦!");
                    key.cancel();
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }

    }

    private void acceptHandler(SelectionKey key){
        ServerSocketChannel server = (ServerSocketChannel)key.channel();
        try {
            SocketChannel client = server.accept();
            if (client == null ) return;
            client.configureBlocking(false);
            System.out.println("-------------------------------------------");
            System.out.println(Thread.currentThread().getName() + " 新客户端：" + client.getRemoteAddress());
            System.out.println("-------------------------------------------");
            // choose a selector and register
            stg.nextSelectorV3(client);

        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    private void printBufferMessage(ByteBuffer buffer){
        System.out.println();
        System.out.print("本次消息内容： ");
        while (buffer.hasRemaining()){
            System.out.print(Character.toChars(buffer.get()));
        }
        System.out.println();
    }
}

